/*
 * Copyright (c) 2007 The Trustees of Indiana University. All rights reserved.
 * IU CSCI B534 Distributed Systems 2007-02-01 http://www.cs.indiana.edu/classes/b534
 */

package edu.indiana.cs.b534.torrent.impl;

import edu.indiana.cs.b534.torrent.Constants;
import edu.indiana.cs.b534.torrent.InfoDictionary;
import edu.indiana.cs.b534.torrent.StorageManager;
import edu.indiana.cs.b534.torrent.TorrentException;
import edu.indiana.cs.b534.torrent.TorrentMetainfo;
import edu.indiana.cs.b534.torrent.Utils;
import edu.indiana.cs.b534.torrent.context.PeerDescription;
import edu.indiana.cs.b534.torrent.context.TorrentContext;
import edu.indiana.cs.b534.torrent.impl.message.TrackerRequestImpl;
import edu.indiana.cs.b534.torrent.impl.message.TrackerResponseImpl;
import edu.indiana.cs.b534.torrent.message.PeerDictionary;
import edu.indiana.cs.b534.torrent.message.TrackerRequest;
import edu.indiana.cs.b534.torrent.message.TrackerResponse;
import edu.indiana.cs.b534.torrent.struct.PiecePeerMapping;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.Vector;
import java.util.logging.Logger;

/**
 * Manages the download of a single torrentMetaInfo and associated pieces
 */
public class TorrentManager implements PieceManager, Runnable {

    public static final String TORRENT_MANAGER_NAME = "edu.indiana.cs.b534.torrentMetaInfo.impl.TorrentManager";
    private static Logger log = Logger.getLogger(TORRENT_MANAGER_NAME);

    private TorrentContext torrentContext;

    // Shared data structures
    private TorrentMetainfo torrentMetaInfo;

    private InfoCollector infoCollector;

    private String myIpAddress;
    private int myPort;

    private List<PieceInfo> piecesToBeDownloaded = new Vector<PieceInfo>();
    private List<PieceInfo> piecesBeingDownloaded = new Vector<PieceInfo>();
    private Map<Integer, PieceInfo> piecesFinishedDownloading = new Hashtable<Integer, PieceInfo>();

    private byte[] infoHash;
    private BittorrentServer bittorrentServer;
    private Hashtable<String, Boolean> peersJustGotBusy;
    private StorageManager storageManager;
    private String status = TrackerRequest.STARTED;
    private boolean isCompleted = false;
    private TrackerUpdater trackerUpdater;
    private ChokingPolicyManager chokingPolicyManager;
    private boolean running = true;

    public TorrentManager(TorrentContext torrentContext) throws TorrentException {
        try {
            this.torrentContext = torrentContext;
            infoCollector = torrentContext.getInfoCollector();
            chokingPolicyManager = new ChokingPolicyManager(torrentContext);

            init();
            log.info("Torrent Manager Initalized");
        } catch (UnknownHostException e) {
            throw new TorrentException(e);
        } catch (SocketException e) {
            throw new TorrentException(e);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }

    private void init() throws TorrentException, UnknownHostException, SocketException, UnsupportedEncodingException {
        this.torrentMetaInfo = torrentContext.getTorrentMetainfo();

        // we need to compute a hash for the file as it should be included whilst communicating
        // with the tracker.
        initInfoHash();

        myIpAddress = Utils.getIpAddress();

        torrentContext.setTorrentMetainfo(torrentMetaInfo);
        myPort = torrentContext.getPort();

        char[] piecesInTargetFile = torrentMetaInfo.getInfo().getPieces().getValue().toCharArray();
        assert piecesInTargetFile.length % 20 == 0;
        int piecesCount = piecesInTargetFile.length / 20;

        int pieceLength = torrentMetaInfo.getInfo().getPieceLength().getValue();

        int fileLength = torrentMetaInfo.getInfo().getLength().getValue();
        int peiceLengthCalcuatedFromTorrentInfo = (int) Math.ceil(((double) fileLength) / pieceLength);
        if (peiceLengthCalcuatedFromTorrentInfo != piecesCount) {
            throw new TorrentException("Calculated size and pieces does not match");
        }


        torrentContext.setPiecesCount(piecesCount);
        PieceInfo[] pieceInfoData = new PieceInfo[piecesCount];
        for (int i = 0; i < pieceInfoData.length - 1; i++) {
            String hash = new String(piecesInTargetFile, i * 20, 20);
            pieceInfoData[i] = new PieceInfo(i, hash, pieceLength);
            piecesToBeDownloaded.add(pieceInfoData[i]);
        }

        //We need to specially handle the last piece as it might not fit in to complete piece length, 
        //Unless we do treat it specially, the hash check will fail
        int i = pieceInfoData.length - 1;
        String hash = new String(piecesInTargetFile, i * 20, 20);
        //We initalize the piece size to what ever left
        pieceInfoData[i] = new PieceInfo(i, hash, fileLength - i * pieceLength);
        piecesToBeDownloaded.add(pieceInfoData[i]);

        // storage manager is responsible for handling all the persistance operations. For the time being
        // we implemented only a memory based storage manager
        storageManager = new MemoryBasedStorageManager(torrentContext);
        if (torrentContext.isSeed()) {
            storageManager.initAsSeed(torrentContext.getSeedDataFile(), pieceInfoData, fileLength);
        } else {
            storageManager.initialize(pieceInfoData, new File(torrentContext.getDestinationFolder()), fileLength);
        }
        torrentContext.setStorageManager(storageManager);

        bittorrentServer = new BittorrentServer(torrentContext.getPort(), this);
        bittorrentServer.start();
    }

    private void initInfoHash() throws TorrentException {
//    	 get the info value and compute it's hash

        try {
            InfoDictionary info = torrentMetaInfo.getInfo();
            ByteArrayOutputStream infoBytes = new ByteArrayOutputStream();
            info.serialize(infoBytes);
            infoHash = Utils.computeHash(infoBytes.toByteArray());

            torrentContext.setInfoHash(infoHash);

        } catch (NoSuchAlgorithmException e) {
            throw new TorrentException(e);
        } catch (IOException e) {
            throw new TorrentException(e);
        }
    }

    public void run() {
        running = true;
        int errCount = 0;
        try {
            // first of all let's contact the tracker and get peer information and other required
            // information from it.
            System.out.print("[ " + torrentContext.getMyPeerId() + " ] Communicating with tracker .....\n");
            TrackerResponse response = updateFromTracker();

            //This is to give enough time to register all peers to tracker
            Thread.sleep(10000);
            response = updateFromTracker();
            System.out.println("[ " + torrentContext.getMyPeerId() + " ] - Done.");

            // we need to continuously keep track of latest updates from the tracker. Let that be
            // handles by a separate TrackerUpdater class
            trackerUpdater = new TrackerUpdater(this);
//            trackerUpdater.start();

            status = TrackerRequest.EMPTY;

            if (torrentContext.isSeed()) {
                //If I am a seed, I will just keep runing so ther can connect to me. but do not try to do anything else
                while (true) {
                    Thread.sleep(2000);
                }
            } else {
                // let's try to connect to the peers list provided by the tracker to make sure they are
                // still alive ;)

                System.out.println("[ " + torrentContext.getMyPeerId() + " ] -Initiating connections with peers ...");
                // PeerDictionarycreate peer instances for peers in tracker response. Which peers to connect?
                List<PeerDictionary> livePeers = trackerUpdater.findLivePeers(response.getPeers());
                if (livePeers.size() == 0) {
                    throw new TorrentException("No live peers found, can not continue");
                }

                // we won't connect to every peer tracker provided us. Let's connected to only a
                // limited number of peers at once
                List<PeerDictionary> peersToConnect = selectPeersToConnect(livePeers);

                // create thread for each instance and start thread
                for (PeerDictionary peerData : peersToConnect) {
                    if (myIpAddress.equals(peerData.getIP()) && myPort == peerData.getPort()) {
                        //Do not connect to myself
                        log.info("Found myself also in the list provided by the tracker. Skipping me.");
                    } else {
                        try {
                            //It is possible connect to a Peer to falied, we continue nevertherless
                            PeerInstance peer = new PeerInstance(infoHash, peerData.getIP(), peerData.getPort(), this, torrentContext.getMyPeerId());
                            torrentContext.addPeer(peer, false);
                            new Thread(peer).start();
                            log.info("Connected to peer " + peerData);
                        } catch (Throwable e) {
                            log.throwing(getClass().getName(), "run", e);
                        }
                    }
                }

                System.out.println("[ " + torrentContext.getMyPeerId() + " ] - Done.");

                // waiting for handshakes to complete
                Thread.sleep(2 * 1000);

                int count = 5;

                int currentlyBusyPeers = 0;

                long runcount = 0;

                System.out.println("[ " + torrentContext.getMyPeerId() + " ] - Starting to download .....");

                peersJustGotBusy = new Hashtable<String, Boolean>(); // this contains the list of peers just started downloading. don't laugh for the name ;)
                infoCollector.notifyStartAllDownloadsEvent();
                // wait for all pieces to be completed downloading
                while (piecesToBeDownloaded.size() > 0 || piecesBeingDownloaded.size() > 0) {

                    //If there are no connected peers, add them from list
                    if (torrentContext.getPeerDescriptions().size() < Constants.MAX_BUSY_CONNECTIONS) {
                        ArrayList<PeerDictionary> toRemove = new ArrayList<PeerDictionary>();
                        Vector<PeerDictionary> availableNonConnectedPeers = torrentContext.getAvalibleNonConnectedPeers();
                        for (int i = 0; i < Math.min(3, availableNonConnectedPeers.size()); i++) {
                            PeerDictionary apeerData = availableNonConnectedPeers.get(i);
                            PeerInstance peer = new PeerInstance(infoHash, apeerData.getIP(), apeerData.getPort(), this, torrentContext.getMyPeerId());
                            torrentContext.addPeer(peer, false);
                            Thread t = new Thread(peer);
                            t.setDaemon(true);
                            t.start();
                            toRemove.add(availableNonConnectedPeers.get(i));
                        }
                        availableNonConnectedPeers.removeAll(toRemove);
                    }

                    int allowedMaxBusyConnections = torrentContext.getMaxAllowedBusyConnections();
//                    System.out.print(allowedMaxBusyConnections + " ");
                    if (currentlyBusyPeers <= allowedMaxBusyConnections) {
//                        ArrayList<PeerInstance> peers = selectPeersWithMostUsefulPieces(allowedMaxBusyConnections - currentlyBusyPeers);
//                        for (PeerInstance peer : peers) {
//                            for (Integer pieceIndex : peer.getPeerDescription().getPiecesList()) {
//                                PieceInfo pieceInfo = getPieceInfo(pieceIndex);
//                                if (piecesToBeDownloaded.contains(pieceInfo)) {
//                                    synchronized (this) {
//                                        try {
//                                            piecesToBeDownloaded.remove(pieceInfo);
//                                            piecesBeingDownloaded.add(pieceInfo);
//                                            peer.startDownload(pieceIndex);
//
//                                            // inform the info collector that we gonna download
//                                            infoCollector.notifyStartDownloadEvent(peer.getPeerDescription().getPeerId(), pieceIndex);
//                                        } catch (Exception e) {
//                                            piecesToBeDownloaded.add(pieceInfo);
//                                            piecesBeingDownloaded.remove(pieceInfo);
//                                        }
//
//                                    }
//                                    break;
//                                }
//                            }
//                        }
//                    }
                        //We reset just got busy list
                        ///peersJustGotBusy.clear();
                        // 1. select rarest n pieces from the one of the unchoked peers
                        PiecePeerMapping[] peerMappings = selectBestPiecesToDownload(allowedMaxBusyConnections - currentlyBusyPeers);

                        //  2. Now select a peer which is not currently used to download a piece
                        for (PiecePeerMapping peerMapping : peerMappings) {
                            for (String peerId : peerMapping.getPeers()) {
                                if (!peersJustGotBusy.containsKey(peerId)) {
                                    PeerInstance peerInstance = torrentContext.getOutgoingPeerInformation().get(peerId);
                                    if (peerInstance == null) {
                                        peerInstance = torrentContext.getIncomingPeerInformation().get(peerId);
                                    }

                                    // when we come to this point peer instance MUST BE NOT NULL
                                    //We remove the piece from toBeDownloaded and add it to being downloaded
                                    PieceInfo pieceInfo = torrentContext.getStorageManager().getPieceInfo(peerMapping.getPieceIndex());


                                    synchronized (this) {
                                        //TODO This is tempeory for this submission, if piece is laredy being downloaded skip
                                        if (!piecesBeingDownloaded.contains(pieceInfo)
                                                && !storageManager.isPieceCompleted(pieceInfo.getIndex())) {
                                            piecesToBeDownloaded.remove(pieceInfo);
                                            piecesBeingDownloaded.add(pieceInfo);

                                            //Start the download
                                            int pieceIndex = peerMapping.getPieceIndex();
                                            log.info("Asking to download piece number " + pieceIndex + " from the peer " + peerInstance.getID());

                                            // inform the info collector that we gonna download
                                            infoCollector.notifyStartDownloadEvent(peerId, pieceIndex);

                                            peerInstance.startDownload(pieceIndex);
                                            peersJustGotBusy.put(peerId, Boolean.TRUE);
                                            break;
                                        }
                                    }
                                }
                            }
                        }

                    }
                    rescheduleTimedOutRequests();
                    // calculate current downloading peers
                    currentlyBusyPeers = calculateBusyPeerCount();
////                    runcount++;
////                    if(runcount%10 == 0){
                    System.out.println(torrentContext.getMyPeerId() + " has " + currentlyBusyPeers + "peers busy " + torrentContext.getPeerDescriptions().size() + " Connected and has pieces " + piecesToBeDownloaded.size() + "  to be downloaded and " + piecesBeingDownloaded.size() + " being downloaded");
////                    }
//
//                    if (currentlyBusyPeers == 0 && piecesToBeDownloaded.size() <= 2) {
//                        System.out.println("problem");
//                    }
                    if (piecesToBeDownloaded.size() == 0 && piecesBeingDownloaded.size() == 1) {
                        errCount++;
                        if (errCount == 4) {
                            System.out.println(piecesBeingDownloaded);
                        }
                    }

                    if (currentlyBusyPeers == 0 && piecesBeingDownloaded.size() == 1) {
                        System.out.println("problem1");
                    }
                    Thread.sleep(1000);
                }
                isCompleted = true;
                System.out.println("************* Download Completed *****************");

                infoCollector.printSummary();

                status = TrackerRequest.COMPLETED;
                updateFromTracker();
                storageManager.setDownloadCompleted();

                // how long to seed? for ever for now
                log.info("Starting to become a full time seed ....");
                while (running) {
                    Thread.sleep(2 * 1000);
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            status = TrackerRequest.STOPPED;
            try {
                updateFromTracker();
            } catch (TorrentException e1) {
                // TODO Auto-generated catch block
                e1.printStackTrace();
            }
        }


    }


    private int calculateBusyPeerCount() {

        int currentlyBusyPeers = 0;
        for (PeerInstance peerInstance : torrentContext.getIncomingPeerInformation().values()) {
            if (peerInstance.isBusy()) {
                currentlyBusyPeers++;
            }
        }
        for (PeerInstance peerInstance : torrentContext.getOutgoingPeerInformation().values()) {
            if (peerInstance.isBusy()) {
                currentlyBusyPeers++;
            }
        }
        return currentlyBusyPeers;
    }

    private PiecePeerMapping[] selectBestPiecesToDownload(int count) throws TorrentException {

        int pieceCount = torrentContext.getPiecesCount();

        PiecePeerMapping[] pieceMappingList = new PiecePeerMapping[pieceCount];

        for (int i = 0; i < pieceMappingList.length; i++) {
            pieceMappingList[i] = new PiecePeerMapping(i);
        }

        // get the unchoked lists of peers
        Collection<PeerDescription> peerDescriptions = torrentContext.getPeerDescriptions();
        synchronized (peerDescriptions) {
            for (PeerDescription peerDescription : peerDescriptions) {
                if (!peerDescription.isChoked()) {

                    String peerId = peerDescription.getPeerId();
                    PeerInstance incomingPeerInstance = torrentContext.getIncomingPeerInformation().get(peerId);
                    PeerInstance outgoingPeerInstance = torrentContext.getOutgoingPeerInformation().get(peerId);

                    if (incomingPeerInstance == null && outgoingPeerInstance == null) {
                        throw new TorrentException("Both Incoming peer instance and out going peer instance can not null at the same time for peer " + peerId);
                    }

                    //TODO we can not iterate over
                    synchronized (peerDescription.getPiecesList()) {
                        for (Integer pieceIndex : peerDescription.getPiecesList()) {

                            // if the piece is not already downloaded and the peer instance is not busy, let's
                            // add it to the mapping
                            if (!piecesFinishedDownloading.containsKey(pieceIndex) && ((incomingPeerInstance != null && !incomingPeerInstance.isBusy())
                                    || (outgoingPeerInstance != null && !outgoingPeerInstance.isBusy()))) {
                                pieceMappingList[pieceIndex].addPeer(peerDescription.getPeerId());
                            }
                        }
                    }
                }
            }
        }

        // select the rarest pieces from them
        Arrays.sort(pieceMappingList);

        // now this piece mapping list contains, pieces that have no peers as well. We need to remove them
        // since this list is already sorted, let's find the index which has no peers for a piece and
        // create a new list from that point forward

        int index;
        for (index = 0; index < pieceMappingList.length; index++) {
            PiecePeerMapping piecePeerMapping = pieceMappingList[index];
            if (piecePeerMapping.getPeers().size() != 0) {
                break;
            }
        }

        PiecePeerMapping[] pieceMappingListWithoutZeros;
        if (count > (pieceCount - index)) {
            pieceMappingListWithoutZeros = new PiecePeerMapping[pieceCount - index];
            System.arraycopy(pieceMappingList, index, pieceMappingListWithoutZeros, 0, pieceCount - index);
        } else {
            pieceMappingListWithoutZeros = new PiecePeerMapping[count];
            System.arraycopy(pieceMappingList, index, pieceMappingListWithoutZeros, 0, count);
        }

        // return the piece -> best peer mapping

        return pieceMappingListWithoutZeros;
    }

    /**
     * Decide which peers to connect at the initialization
     *
     * @param peers
     * @return
     */

    private List<PeerDictionary> selectPeersToConnect(List<PeerDictionary> peers) {
        if (peers.size() > Constants.MAX_BUSY_CONNECTIONS) {
            List<PeerDictionary> selectedPeers = new ArrayList<PeerDictionary>();
            for (int i = 0; i < Constants.MAX_BUSY_CONNECTIONS; i++) {
                selectedPeers.add(peers.get(i));
            }
            return selectedPeers;
        } else {
            return peers;
        }
    }

    public synchronized TrackerResponse updateFromTracker() throws TorrentException {
        try {
            TrackerRequest trackerRequest;
            int downloaded = storageManager.getTotalDownloadedBytes();
            int uploaded = storageManager.getTotalUploadedBytes();
            int left = storageManager.getBytesLeftToBeDownloaded();
            trackerRequest = new TrackerRequestImpl(downloaded, status, myIpAddress,
                    infoHash, left, torrentContext.getMyPeerId(), myPort, uploaded);

//            if (torrentContext.isSeed()) {
//            } else {
//                trackerRequest = new TrackerRequestImpl(0, TrackerRequest.STARTED, myIpAddress,
//                        infoHash, torrentMetaInfo.getInfo().getLength().getValue(), torrentContext.getMyPeerId(), myPort, 0);
//            }
            // create and call tracker with 'tracker request' message
            HTTPCommunicationManager HTTPCommunicationManager = new HTTPCommunicationManager();
            String trackerURL = torrentMetaInfo.getAnnounce().getValue() + "?" + trackerRequest.getQueryString();
            log.info("tracker URL = " + trackerURL);
            InputStream in = HTTPCommunicationManager.sendReceive(trackerURL);
            return new TrackerResponseImpl(in);


        } catch (IOException e) {
            throw new TorrentException(e);
        }
    }

//    public PieceInfo getPieceToDownload(byte[] availableBitField) {
//        /* TODO */
//        // remove and return a piece from 'to be downloaded' list. How do you do the selection?
//        // move piece to 'being downloaded' list
//        // sync access?
//        return null;
//    }

    public boolean setPieceDownloaded(PieceInfo pieceInfo, String peerId) throws TorrentException {
        Set<PeerInstance> peers = new HashSet<PeerInstance>();
        peers.addAll(torrentContext.getIncomingPeerInformation().values());
        peers.addAll(torrentContext.getOutgoingPeerInformation().values());
        for (PeerInstance peer : peers) {
            peer.sendHaveMessage(pieceInfo);
        }
        synchronized (this) {
            piecesBeingDownloaded.remove(pieceInfo);
            piecesFinishedDownloading.put(pieceInfo.getIndex(), pieceInfo);
            peersJustGotBusy.remove(peerId);
        }

        log.severe("Completed downloading the piece " + pieceInfo.getIndex());
        infoCollector.setPieceDownloadedFromPeer(peerId, pieceInfo.getIndex());
        return true;
    }

    public void cancelPieceDownload(PieceInfo piece, PeerInstance peerInstance) {
        if (piece == null || peerInstance == null) {
            throw new RuntimeException("When cancel piece Download called, peer instance " + peerInstance + "piece " + piece);
        }
// cancel a piece that was flagged as being downloaded.
// Move piece from 'being downloaded' list to 'to be downloaded' list
        synchronized (this) {
            piecesBeingDownloaded.remove(piece);
            piecesToBeDownloaded.add(piece);
            peersJustGotBusy.remove(peerInstance.getID());
        }
    }

    /**
     * @param index
     * @return
     */
    public synchronized PieceInfo getPieceInfo(int index) {
        //sync accsess
        return torrentContext.getStorageManager().getPieceInfo(index);
    }

    public boolean interestedMessageReceived(PeerInstance peerInstance) {

//        if (torrentContext.isSeed()) return true;
//
//        boolean chokingPolicyDecision = !chokingPolicyManager.shouldIChoke(peerInstance);
//
//        System.out.println("Choking Policy Decision of " + torrentContext.getMyPeerId() + " for peer " + peerInstance.getID() + " = " + chokingPolicyDecision);
        return true;
    }

    public TorrentContext getTorrentContext() {
        return torrentContext;
    }


    public boolean isCompleted() {
        return isCompleted;
    }

    public void rescheduleTimedOutRequests() {
        Collection<PeerDescription> peers = torrentContext.getPeerDescriptions();
        for (PeerDescription peer : peers) {
            if (peer.isSentInterested() && peer.isChoked()) {
                long timeEplipsed = System.currentTimeMillis() - peer.getTimeLastRequestOrInterstedSent();
                if (timeEplipsed > Constants.TIME_OUT_FOR_RESPONSE) {
                    //Some Intersted message has gone unanswered, retry
                    System.out.println("Intersted Message from " + torrentContext.getMyPeerId() + " to " + peer.getPeerId() + " has falied");
                    PeerInstance peerInstance = Utils.findPeerInstance(torrentContext, peer.getPeerId());
                    if (peer.getRetryCout() == 0) {
                        try {
                            peerInstance.sendInterested();
                        } catch (TorrentException e) {
                            log.throwing(getClass().getName(), "sendInterested", e);
                        }
                    } else {
                        //TODO What to do?
                        peerInstance.shutDownConnectionToPeer();
                    }
                }
            } else if (peer.getCurrentDownloadingPiece() != -1) {
                long timeEplipsed = System.currentTimeMillis() - peer.getTimeLastRequestOrInterstedSent();
                if (timeEplipsed > Constants.TIME_OUT_FOR_RESPONSE) {
                    synchronized (this) {
                        PeerInstance peerInstance = Utils.findPeerInstance(torrentContext, peer.getPeerId());
                        cancelPieceDownload(storageManager.getPieceInfo(peer.getCurrentDownloadingPiece()), peerInstance);
                        peerInstance.resetBusyState();
                    }
                    //TODO either shutdown the PeerInstance or reset it
                    System.out.println("Request Message from " + torrentContext.getMyPeerId() + " to " + peer.getPeerId() + " has timed out");
                }
            }

        }

    }

    private ArrayList<PeerInstance> selectPeersWithMostUsefulPieces(int n) throws TorrentException {
        Collection<PeerDescription> peers = torrentContext.getPeerDescriptions();
        PeerBean[] peerUsefulPieceCount = new PeerBean[peers.size()];
        int index = 0;
        synchronized (peers) {
            for (PeerDescription peer : peers) {
                peerUsefulPieceCount[index] = new PeerBean(peer);
                for (Integer piece : peer.getPiecesList()) {
                    if (piecesToBeDownloaded.contains(this.getPieceInfo(piece.intValue()))) {
                        peerUsefulPieceCount[index].freq += 1;
                    }
                }
                index++;
            }
        }
        Arrays.sort(peerUsefulPieceCount);

        ArrayList<PeerInstance> results = new ArrayList<PeerInstance>();

        int slectedPeerCount = 0;
        for (int i = 0; i < peerUsefulPieceCount.length && slectedPeerCount < n; i++) {
            PeerInstance instance = torrentContext.getIncomingPeerInformation().get(peerUsefulPieceCount[i].peer.getPeerId());
            if (instance == null) {
                instance = torrentContext.getOutgoingPeerInformation().get(peerUsefulPieceCount[i].peer.getPeerId());
            }
            if (instance == null) {
                throw new TorrentException("Can not  find peer instance for " + peerUsefulPieceCount[i].peer.getPeerId());
            }
            if (!instance.isBusy()) {
                results.add(instance);
                slectedPeerCount++;
            }
            if (slectedPeerCount == n) {
                break;
            }
        }
        return results;

    }

    class PeerBean implements Comparable<PeerBean> {

        int freq = 0;
        PeerDescription peer;

        public PeerBean(PeerDescription peer) {
            super();
            this.peer = peer;
        }

        public int compareTo(PeerBean o) {
            return o.freq - freq;
        }
    }


    public void shutDown() throws TorrentException {
        running = false;
        this.status = TrackerRequest.STOPPED;
        updateFromTracker();
    }


    public static void main(String[] args) {
        try {
            Properties p = new Properties();
            FileInputStream in = new FileInputStream(args[0]);
            p.load(in);
            in.close();

            TorrentManager manager = new TorrentManager(new TorrentContext(p));
            manager.run();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}
